/**
 * Copyright 2006-2015 handu.com
 * <p/>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.iepacj.mq.rabbit.demo1;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消费者
 */
public class MessageReceiver {

    private static String QUEUE_NAME = "queue_demo1";

    private static String EXCHANGE = "first_exchange";

    private static final Logger logger = LoggerFactory.getLogger(MessageReceiver.class);

    private static ExecutorService es = Executors.newFixedThreadPool(100);


    public static void main(String[] args) throws Exception {

        ConnectionFactory connFact = new ConnectionFactory();
        final Connection conn = connFact.newConnection();

        final MessageReceiver rece = new MessageReceiver();

        for (int i = 2; i < 3; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        rece.consume(conn);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "consumer-" + i).start();}

    }

    public void consume(Connection conn) throws Exception{
        final Channel channel = conn.createChannel();
        //final Channel channel1 = conn.createChannel();
        channel.exchangeDeclare(EXCHANGE, "direct");
        AMQP.Queue.DeclareOk declareOk = channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //AMQP.Queue.DeclareOk declareOk1 = channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE, QUEUE_NAME);
        //channel1.queueBind(QUEUE_NAME, EXCHANGE, QUEUE_NAME);

        channel.basicQos(1);


        logger.info("ConsumerCount::{}", declareOk.getConsumerCount());
        logger.info("getQueue::{}", declareOk.getQueue());
        logger.info("protocolMethodName::{}", declareOk.protocolMethodName());
        logger.info(" {}", declareOk.getMessageCount());

        logger.info("消费者准备接受消息::{}", QUEUE_NAME);

        final QueueingConsumer consumer = new QueueingConsumer(channel);
        String s = channel.basicConsume(QUEUE_NAME, false, consumer);
        //String s2 = channel1.basicConsume(QUEUE_NAME, true, consumer);

        logger.info("s::{}::{}", s, "");

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            es.execute(new Target(delivery, channel));
        }
    }

    public void handleMessage(QueueingConsumer.Delivery delivery, Channel channel) throws Exception {
        String msg = new String(delivery.getBody());

        logger.info("又收到消息::{}", msg);

        //AMQP.BasicProperties prop = delivery.getProperties();
        //Map<String, Object> headers = prop.getHeaders();
        //String msgId = prop.getMessageId();
        //logger.info("header::{}::messageId::{}", headers, msgId);

        // 模拟耗时
        Thread.sleep(6000);

        //logger.info("****************处理消息结束, 发送ACK*****************");

        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }

    class Target implements Runnable {

        private QueueingConsumer.Delivery delivery;

        private Channel channel;

        public Target(QueueingConsumer.Delivery delivery, Channel channel) {
            this.delivery = delivery;
            this.channel = channel;
        }

        @Override
        public void run() {
            try {
                handleMessage(delivery, channel);
            } catch (Exception e) {
            }
        }
    }
}












































